package eval;

import com.google.common.collect.Queues;
import network.Device;
import network.Network;
import network.Rule;
import networkLoader.FBNetwork;
import partition.SubgraphPartition;
import partition.SubspacePartition;

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

public class Dispatcher implements Runnable {
    private BlockingQueue<Rule> blockingQueue;
    private HashMap<Long, BlockingDeque<Rule>> subspaceToBlockingQueue;
    private HashMap<Long, SubspacePartition> idToSubspace;

    public Dispatcher(BlockingQueue<Rule> blockingQueue, HashMap<Long, SubspacePartition> idToSubspace) {
        this.blockingQueue = blockingQueue;
        this.idToSubspace = idToSubspace;
    }

    public void run() {
        List<Rule> rules = new ArrayList<>();
        while (true) {
            try {
//                int n = Queues.drain(blockingQueue, rules, Integer.MAX_VALUE, 0, TimeUnit.SECONDS);
                rules.add(blockingQueue.take());
                int n = 1;
                n += blockingQueue.drainTo(rules);
                boolean containClose = false;
                HashSet<Device> newClosed = new HashSet<>();
                for (int i = 0; i < n; i++) {
                    Rule rule = rules.get(rules.size() - 1 - i);
                    if (rule.isLast()) {
                        containClose = true;
                        newClosed.add(rule.getDevice());
                    }
                }
                if (containClose) {
//                    System.out.println("got new closed " + newClosed.size());
                    dispatch(rules, newClosed);
                    rules.clear();
//                    if (closed.size() == 6016) return;
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static int processedRules = 0;
    private void dispatch(List<Rule> rules, HashSet<Device> newClosed) throws InterruptedException {
        Map<Long, List<Rule>> rulesList = rules.stream().collect(Collectors.groupingBy(rule -> (rule.getMatch().longValue() >> 24)));
        for (Map.Entry<Long, SubspacePartition> entry : idToSubspace.entrySet()) {
            List<Rule> rulesInSubspace = rulesList.get(entry.getKey()) != null ? rulesList.get(entry.getKey()) : new ArrayList<>();
            Batch batch = new Batch(rulesInSubspace, newClosed);
            entry.getValue().getPendingBatches().addAll(batch.getRules());
        }
        processedRules += rules.size();
    }
}
